package com.sunny.client.discover;

import com.sunny.client.loadbalance.LoadBalanceStrategy;
import com.sunny.client.loadbalance.RandomLoadBalance;
import com.sunny.service.ZKConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * 服务发现
 *
 * @author Sunny
 * @date 2020/2/16
 */
public class ZKServiceDiscovery implements IServiceDiscovery {

    private CuratorFramework curator;

    private List<String> serviceAddress = new ArrayList<>(); //服务地址的本地缓存

    {
        //初始化zookeeper的连接， 会话超时时间是3s，衰减重试
        curator = CuratorFrameworkFactory.builder().
                connectString(ZKConfig.CONNECT_STRING).sessionTimeoutMs(3000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).
                namespace(ZKConfig.NAMESPACE)
                .build();
        curator.start();
    }

    @Override
    public String discover(String serviceName) {
        String servicePath = "/" + serviceName;
        try {
            if (serviceAddress.isEmpty()) {
                serviceAddress = curator.getChildren().forPath(servicePath);
                watcher(servicePath);
            }

            LoadBalanceStrategy loadBalanceStrategy = new RandomLoadBalance();
            String host = loadBalanceStrategy.selectHost(serviceAddress);
            if (StringUtils.isEmpty(host)) {
                throw new RuntimeException("discovery service exception! address not found!");
            }
            return host;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 监听 path节点, 如果子节点有变化, 重新赋值服务地址List
     *
     * @param path 服务path
     * @throws Exception
     */
    private void watcher(final String path) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curator, path, true);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("watcher callback, type: " + pathChildrenCacheEvent.getType());
                serviceAddress = curatorFramework.getChildren().forPath(path);
            }
        };
        pathChildrenCache.getListenable().addListener(listener);
        pathChildrenCache.start();
    }
}
